//package com.hcy.schedulingapi.task;
//
//import com.hcy.common.dto.response.GaoDeGfidsResponseDTO;
//import com.hcy.controller.LocationInfoClient;
//import com.hcy.driver.DriverBaseInfoClient;
//import com.hcy.mqclient.message.MessageServiceClient;
//import com.hcy.rabbitbasicmodel.enums.Code;
//import com.hcy.rabbitbasicmodel.model.ResponseData;
//import com.hcy.schedulingapi.service.mqconsumer.MQDispatcherInit;
//import com.hcy.schedulingapi.service.mqconsumer.MQDispatcherOrderConsumer;
//import com.hcy.schedulingapi.utils.RedisUtils;
//import org.apache.commons.lang.builder.ToStringBuilder;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.ApplicationArguments;
//import org.springframework.boot.ApplicationRunner;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//
///**
// * @Author huchenying
// * @Description
// * @Date 2022/3/22 22:11
// */
//@Component
//public class ConsumerInit implements ApplicationRunner {
//
//    Logger logger = LoggerFactory.getLogger(ConsumerInit.class);
//
//    @Autowired
//    private LocationInfoClient locationInfoClient;
//
//    @Value("${MQADDR}")
//    private String MQADDR;
//    @Autowired
//    private MQDispatcherOrderConsumer mqDispatcherOrderConsumer;
//
//    @Autowired
//    RedisUtils redisUtils;
//
//    @Autowired
//    MessageServiceClient messageServiceClient;
//
//    @Autowired
//    DriverBaseInfoClient driverBaseInfoClient;
//
//    @Override
//    public void run(ApplicationArguments args) throws Exception {
//
//        logger.info("初始化派单消费者 start ...");
//        ResponseData<List<GaoDeGfidsResponseDTO>> allTopic = locationInfoClient.getAllTopic();
//        if(allTopic.getCode() != Code.Success){
//            logger.error("获取Topic配置失败，初始化派单消费者 终止 ...");
//        }
//        if(allTopic.getCode() == Code.Success){
//            logger.info("获取到配置信息开始初始化");
//            List<GaoDeGfidsResponseDTO> data = allTopic.getData();
//            for (GaoDeGfidsResponseDTO datum : data) {
//                datum.setAddress(MQADDR);
//                createConsumer(datum);
//            }
//            logger.info("初始化完成");
//        }else {
//            logger.error("初始化派单消费队列为空！");
//        }
//    }
//
//    public void createConsumer(GaoDeGfidsResponseDTO gaoDeGfid){
//        logger.info("开始创建地区consumer param -> {}", ToStringBuilder.reflectionToString(gaoDeGfid));
//
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(gaoDeGfid.getGfid().toString());
//        //设置NameServer地址
//        consumer.setNamesrvAddr(gaoDeGfid.getAddress());
//        //设置起始消费位置
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//        //设置消费超时时间
//        consumer.setConsumeTimeout(10 * 60 * 1000);
//        //指定消费Topic和Tag
//        try {
//            consumer.subscribe(gaoDeGfid.getAddress(), "*");
//        } catch (MQClientException e) {
//            throw new RuntimeException("设置topic异常");
//        }
//
//        //指定消费模式 默认集群模式
//        consumer.setMessageModel(MessageModel.CLUSTERING);
//        //设置消费启动线程
//        consumer.setConsumeThreadMin(1);
//        consumer.setConsumeThreadMax(1);
//        //设置每批消费多少
//        consumer.setConsumeMessageBatchMaxSize(1);
//        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//
//            //消费一条消息
//            logger.info("msgs size = [{}]", msgs.size());
//            for (MessageExt msg : msgs) {
////                System.out.println(msg.getKeys());
//                try {
//                    consumer(msg);
//                    logger.info("消费成功！");
//                } catch (InterruptedException e) {
//                    System.out.println(e);
//                    e.printStackTrace();
//                    logger.error("消费失败");
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                }
//            }
//            //返回消费状态
//            //重新消费
////                ConsumeConcurrentlyStatus.RECONSUME_LATER
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
//        //开启消费
//        try {
//            consumer.start();
//            logger.info("消费派单consumer启动成功！");
//        } catch (MQClientException e) {
//            e.printStackTrace();
//            logger.error("消费派单consumer启动失败！");
//        }
//
//    }
//}
